package com.shengzai.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo3Window {
  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()

    conf.setMaster("local[2]")

    conf.setAppName("window")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    val rTds: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.146.136", 8888)

    val mapDS: DStream[(String, Int)] = rTds.flatMap(
      line => {
        line.split(",")
      }
    ).map(
      line => {
        (line, 1)
      }
    )

    val windowFun: DStream[(String, Int)] = mapDS.reduceByKeyAndWindow(
      (x, y) => x + y,
      //取前15秒的数据
      windowDuration = Durations.seconds(15),
      //5秒钟滑动一次
      slideDuration = Durations.seconds(5)
    )

    windowFun.print()
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()



  }

}
